package com.zzl.spark.sql.hive;

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

public class HiveDataSource {

    public static void main(String[] args) {
        SparkSession hiveDataSource = SparkSession.builder().appName("HiveDataSource").master("local[2]").enableHiveSupport().getOrCreate();

        hiveDataSource.sql("drop table if exists student_infos");

        hiveDataSource.sql("create table if not exists student_infos(name STRING,age INT) row format delimited fields terminated by '\t' ");

        hiveDataSource.sql("LOAD DATA" +
                " LOCAL INPATH '/root/resource/student_infos' " +
                "INTO TABLE student_infos");

        hiveDataSource.sql("drop table if exists student_scores");

        hiveDataSource.sql("create table if not exists student_scores(name STRING,score INT) row format delimited fields terminated by '\t' ");

        hiveDataSource.sql("LOAD DATA" +
                " LOCAL INPATH '/root/resource/student_scores' " +
                "INTO TABLE student_scores");

        Dataset<Row> goodStudentsDs = hiveDataSource.sql("select si.*,ss.score from student_infos si join student_scores ss on si.name=ss.name where ss.score>=80");

        hiveDataSource.sql("drop table if exists good_student_infos");

        goodStudentsDs.write().saveAsTable("good_student_infos");

        Dataset<Row> good_student_infos = hiveDataSource.table("good_student_infos");
        good_student_infos.show();
        hiveDataSource.stop();

    }
}
